package com.sixeco.apigetway.common;

import java.net.SocketAddress;
import java.security.SecureRandom;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.sixeco.apigetway.handle.BcHttpHandle;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.util.AttributeKey;
import io.netty.util.concurrent.EventExecutor;

/**
 * 
 * @author tianzhenjiu
 *
 */
public class CommonIO {

	Logger logger = LoggerFactory.getLogger(getClass());

	/**
	 * 标志是否初始化了channnel
	 */
	volatile boolean isInit = false;

	private static CommonIO commonIO = new CommonIO();

	final Bootstrap httpbootstrap = new Bootstrap();

	final ThreadLocal<Set<Channel>> localChannels = new ThreadLocal<>();

	int initChannelCount = CommonVar.Proxy.initchannelCount;

	AtomicInteger bcChannels = new AtomicInteger(0);

	/**
	 * 重连
	 * 
	 * @param oldChannel 旧的channel
	 * @param host       主机地址
	 * @param port       端口
	 * @return
	 */
	public void reconnection0(Channel oldChannel, SocketAddress socketAddress) {

		Set<Channel> activeChannels=localChannels.get();
		
		if (activeChannels.contains(oldChannel)) {
			activeChannels.remove(oldChannel);
			bcChannels.decrementAndGet();
		}

		AttributeKey<Channel> bcChannelAttr = CommonVar.Proxy.bcChannelAttr;
		AttributeKey<Channel> abChannelAttr = CommonVar.Proxy.abChannelAttr;

		if (oldChannel.hasAttr(abChannelAttr)) {

			/**
			 * 如果前置channel还活着,后置不活了必须重连
			 */
			Channel abChannel = oldChannel.attr(abChannelAttr).get();
			if (abChannel != null && abChannel.isActive()) {
				ChannelFuture channelFuture = connection0(socketAddress);

				channelFuture.addListener((listener) -> {

					if (listener.isSuccess()) {

						/**
						 * 把原有channel绑定
						 */
						Channel channel = channelFuture.channel();
						abChannel.attr(bcChannelAttr).set(channel);
						channel.attr(abChannelAttr).set(abChannel);
					}

				});

				return;
			}
		}

		/**
		 * 如果可用连接数量小于最小数量 必须重连
		 */
		
		int activeSize=activeChannels.size();
		if(activeSize<CommonVar.Proxy.initchannelCount) {			
			connection0(randomSocket());
			logger.debug("连接数{}少了{}必须重连", activeSize, CommonVar.Proxy.initchannelCount);
		}
	

	}

	/**
	 * 连接一条链路
	 * 
	 * @param host
	 * @param port
	 * @return
	 */
	private ChannelFuture connection0(SocketAddress socketAddress) {

		Bootstrap bootstrap = httpbootstrap;
		
		Set<Channel> activeChannels=localChannels.get();
		ChannelFuture channelFuture = bootstrap.connect(socketAddress);

		channelFuture.addListener((listener) -> {

			if (listener.isSuccess()) {
				activeChannels.add(channelFuture.channel());
				
				logger.debug("添加channel->{},总共连接数{}", channelFuture.channel().remoteAddress(),bcChannels.incrementAndGet());

			} else {
				logger.warn("{}连接失败{}", listener.cause(),socketAddress);
			}
		});

		return channelFuture;

	}

	/**
	 * 初始化配置后端连接
	 */
	public void initBcChannel() {

		if (isInit) {
			return;
		}

		
		

		NioEventLoopGroup httpgroup = CommonVar.Proxy.httpgroup;

		Bootstrap bootstrap = httpbootstrap;
		bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS,3000);
		bootstrap.option(ChannelOption.TCP_NODELAY, true);
		bootstrap.group(httpgroup).channel(NioSocketChannel.class)
				.handler(new ChannelInitializer<SocketChannel>() {
					@Override
					protected void initChannel(SocketChannel ch) throws Exception {

						ChannelPipeline channelPipeline = ch.pipeline();
						
						channelPipeline.addLast(new HttpClientCodec());
						channelPipeline.addLast(new ChunkedWriteHandler());
						channelPipeline.addLast(new HttpObjectAggregator(819200));

						channelPipeline.addLast(new BcHttpHandle(CommonIO.this::reconnection0));
						
					}

				});
		

		Iterator<EventExecutor> iterator=httpgroup.iterator();
		
		while(iterator.hasNext()) {
			
			iterator.next().execute(() -> {
				Set<Channel> channels = new HashSet<>();
				localChannels.set(channels);

				for (int i = 0; i < initChannelCount; i++) {
					connection0(randomSocket());
				}
			});
		}
		
	

		isInit = true;

	}

	/**
	 * 绑定后置channel并且把内容写到后置channel
	 * 
	 * @param requestChannel 前置channel
	 * @param writeTimes 第几次写入
	 * @param msg 消息
	 */
	public void bindAndWriteBcActiveChannel(Channel requestChannel,int writeTimes, FullHttpRequest request) {

		request.retain();//1->2
		
		
		Channel activeChannel= lookUpBindBcChannelActive(requestChannel);
		if (activeChannel != null && activeChannel.isActive()) {
			activeChannel.writeAndFlush(request).addListener(new WriteProcessListener( activeChannel, writeTimes, request));
		} else {
			bindWriteNewChannel(requestChannel,writeTimes, request);

			logger.debug("没有找到需要新建后置channel->{}->{}", requestChannel, activeChannel);

		}

	}

	/***
	 * 绑定新的前置和后置channel
	 * 
	 * @param requestChannel 前置channel
	 */
	void bindWriteNewChannel(Channel requestChannel, int writeTime,FullHttpRequest request) {

		AttributeKey<Channel> bcChannelAttr = CommonVar.Proxy.bcChannelAttr;

		AttributeKey<Channel> abChannelAttr = CommonVar.Proxy.abChannelAttr;

		ChannelFuture channelFuture = connection0(randomSocket());
		channelFuture.addListener((listener) -> {

			if (listener.isSuccess()) {
				Channel channel = channelFuture.channel();

				requestChannel.attr(bcChannelAttr).set(channel);
				channel.attr(abChannelAttr).set(requestChannel);

				channel.writeAndFlush(request).addListener(new WriteProcessListener(channel, writeTime, request));
			} else {
				CommonIO.getInstance().writeMsgAndClose(requestChannel, CommonVar.createServerErrorRequest());
				request.release();
			}
		});

	}

	/**
	 * 寻找后置channel
	 * 
	 * @param requestChannel
	 * @return
	 */
	Channel lookUpBindBcChannelActive(Channel requestChannel) {

		AttributeKey<Channel> abChannelAttr = CommonVar.Proxy.abChannelAttr;

		AttributeKey<Channel> bcChannelAttr = CommonVar.Proxy.bcChannelAttr;

		Channel activeChannel = null;

		if (requestChannel.hasAttr(bcChannelAttr)) {
			activeChannel = requestChannel.attr(bcChannelAttr).get();
			if (activeChannel != null && activeChannel.isActive()) {

				logger.debug("直接使用后置存活channel");
				return activeChannel;
			}
		}

		Set<Channel> channels=localChannels.get();
		
		logger.info("channels->{}",channels);
		if(channels==null) {
			return null;
		}
		for (Channel channel : channels) {
			/**
			 * 如果后置channe不可用了直接略过
			 */
			if (!channel.isActive()) {
				continue;
			}

			/**
			 * 保证后置channel与前置channel都是在同一个线程里面
			 */
			if (!channel.eventLoop().inEventLoop()) {
				logger.debug("channel为什么会与当前线程不一致?{}",channel);
				continue;
			}

			/**
			 * 如果后置channel没有这个属性可用直接使用
			 */
			if (!channel.hasAttr(abChannelAttr)) {

				logger.debug("直接使用后置连接池没有绑定的channel");
				activeChannel = channel;
				break;
			}

			/**
			 * 如果后置channel中的对应的前置channel不可用也可用直接使用
			 */
			if (!channel.attr(abChannelAttr).get().isActive()) {
				activeChannel = channel;
				logger.debug("直接使用后置连接池绑定无效的并且一个线程的channel");
				break;

			}

		}

		if (activeChannel != null && activeChannel.isActive()) {
			requestChannel.attr(bcChannelAttr).set(activeChannel);
			activeChannel.attr(abChannelAttr).set(requestChannel);

		}

		return activeChannel;
	}

	/**
	 * 随机取出一个socket
	 * @return
	 */
	SocketAddress randomSocket() {
		
		
		List<SocketAddress> addresses=CommonVar.Proxy.proxySockets;
		if(addresses==null) {
			return null;
		}
		
		SecureRandom random=new SecureRandom();
		
		int size=addresses.size();
		if(size<=0) {
			logger.warn("没有可用的socket");
			 System.exit(0);
		}
		int index=random.nextInt(size);
		if(index>=size) {
			index=0;
		}
	
		return addresses.get(index);
		
	}
	/**
	 * 对channel写消息并且关闭掉
	 * 
	 * @param channel
	 * @param msg
	 * @return
	 */
	public ChannelFuture writeMsgAndClose(Channel channel, Object msg) {

		ChannelFuture channelFuture = channel.writeAndFlush(msg).addListener(ChannelFutureListener.CLOSE);

		return channelFuture;
	}

	private CommonIO() {
		initBcChannel();
	}

	/**
	 * 获取单利对象
	 * 
	 * @return
	 */
	public static CommonIO getInstance() {

		return commonIO;
	}
}
